package wjl.project.dubbotestshop;

import lombok.extern.log4j.Log4j2;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;




/**
 * GoodsService
 *
 *  J
 *
 *  在一个分布式系统中，多台机器上部署了多个服务，
 *  当客户端一个用户发起一个数据插入请求时，如果没有分布式锁机制保证，
 *  那么那多台机器上的多个服务可能进行并发插入操作，导致数据重复插入，
 *
 *
 * @blame Android Team
 */
@Service
@Log4j2
public class GoodsService {
    @Resource
    GoodsRepository goodsRepository;

    @Resource
    KafkaSender kafkaSender;

    private static int num = 0;

    @Resource
    RedisTemplate<String,String> redisTemplate;
    private static Map<Long,Integer>  goodsMapNum = new ConcurrentHashMap<>(8);

    List<Goods> getGoodsList(){
        return goodsRepository.findAll();
    }

    public boolean buyGoodsSync(Long goodsId){
//        服务器缓存
        if (goodsMapNum.containsKey(goodsId)){
            if (goodsMapNum.get(goodsId)==0){
//                log.info("服务器缓存检测到商品已经卖完");
                return false;
            }
        }
//        redis 自减操作
        try {
            Long remain =  redisTemplate.opsForValue().decrement(DubboTestShopApplication.MIAOSHA_PREFIX+goodsId);
//            log.info("剩余数量为："+remain);
            if (remain == null || remain < 0){
                goodsMapNum.put(goodsId,0);
//                log.info("redis缓存检测到商品已经卖完");
                return false;
            }
            Integer remain_int = Math.toIntExact(remain);
            goodsMapNum.put(goodsId,remain_int);
            Goods goods = Goods.builder().goodsId(goodsId).goodsNum(remain_int).build();

            ListenableFuture<SendResult<String, String>> future =
                    kafkaSender.BuyGoodsSync(goods);
            //回调函数
            future.addCallback(new ListenableFutureCallback<>() {
                @Override
                public void onFailure(Throwable throwable) {

                    log.error(throwable.getMessage()+"-----Send message failed,绕过消息队列直接写入数据库");
                    goodsRepository.updateGoodsNum(goods.getGoodsNum(),goods.getGoodsId());
                }

                @Override
                public void onSuccess(SendResult<String, String> stringEntitySendResult) {
                    log.info("成功次数:"+num);
                    num++;
                    log.info("Send message success");
                }
            });
            return true;
        }catch (Exception e){
            e.printStackTrace();
            return  false;
        }
    }

    /**
     * 在高并发的条件了，多个线程同时执行 buyGoods方法，及时kafka是有序的， 还是不能保证在进入消息队列前的有序性
     * Java中我们一般可以使用synchronized语法和ReetrantLock去保证，这实际上是本地锁的方式。但是现在公司都是流行分布式架构，在分布式环境下，如何保证不同节点的线程同步执行呢？
     *
     * 这里虽然使用了ConcurrentHashMap 保证了共享资源的同步  但是在分布式环境下，无法保证代码块的同步
     *
     *  释放锁时需要验证value值，也就是说我们在获取锁的时候需要设置一个value，不能直接用del key这种粗暴的方式，因为直接del key任何客户端都可以进行解锁了
     *  所以解锁时，我们需要判断锁是否是自己的，基于value值来判断， 并加上超时时间， 保证不会误解锁
     *
     *  考虑订阅 锁信息
    * */

    public boolean buyGoods(Long goodsId){
        if (goodsMapNum.containsKey(goodsId)){
            if (goodsMapNum.get(goodsId)==0){
                return false;
            }
        }
//        1. 在这里 获取redis 锁  因为加锁和解锁涉及到一个以上的命令，为了保证原子性，所以一般使用lua脚本来执行命令
//        2。为了保证值的唯一性，这里使用UUID来生成随机ID作为值，并将该值存在本地线程变量中，当重入锁或解锁时可以直接从变量中拿以保证相同的key拥有的value一致。
        try {
            Long remain =  redisTemplate.opsForValue().decrement(DubboTestShopApplication.MIAOSHA_PREFIX+goodsId);
            
            if (remain == null || remain < 0){
                goodsMapNum.put(goodsId,0);
                return false;
            }
            Integer remain_int = Math.toIntExact(remain);
            goodsMapNum.put(goodsId,remain_int);
            Goods goods = Goods.builder().goodsId(goodsId).goodsNum(remain_int).build();
            return kafkaSender.BuyGoods(goods);
        }catch (Exception e){
            e.printStackTrace();
            return  false;
        }
    }
}
